package com.litchi.pools;

import lombok.extern.slf4j.Slf4j;

import java.util.concurrent.*;

/**
 * @Author: Jiangxx
 * @Date: 2023/10/18
 * @Description:
 */
@Slf4j
public class ChangeEventThreadPoolHolder {

    public static final String POOL_NAME = "CDC-LISTENER_POOL";

    /**
     * CPU核数
     **/
    public static final int CPU_COUNT = Runtime.getRuntime().availableProcessors();

    /**
     * IO线程池最大线程数
     */
    public static final int IO_MAX = Math.max(2, CPU_COUNT * 2);

    /**
     * 空闲保活时限，单位秒
     */
    public static final int KEEP_ALIVE_SECONDS = 30;
    /**
     * 有界队列size
     */
    public static final int QUEUE_SIZE = 10000;

    public static final ThreadPoolExecutor EXECUTOR = new ThreadPoolExecutor(
            IO_MAX,
            IO_MAX,
            KEEP_ALIVE_SECONDS,
            TimeUnit.SECONDS,
            new LinkedBlockingQueue<>(QUEUE_SIZE),
            new CustomThreadFactory("io"));

    public static ThreadPoolExecutor getInnerExecutor() {
        return EXECUTOR;
    }

    static {
        log.info("线程池已经初始化");

        EXECUTOR.allowCoreThreadTimeOut(true);
        //JVM关闭时的钩子函数
        Runtime.getRuntime().addShutdownHook(
                new ShutdownHookThread("cdc数据处理线程池", (Callable<Void>) () -> {
                    //优雅关闭线程池
                    shutdownThreadPoolGracefully(EXECUTOR);
                    return null;
                }));
    }

    public static void shutdownThreadPoolGracefully(ExecutorService threadPool) {
        if (threadPool == null || threadPool.isTerminated()) {
            return;
        }
        try {
            // 拒绝接受新任务
            threadPool.shutdown();
        } catch (SecurityException | NullPointerException e) {
            return;
        }
        try {
            // 等待 60 s，等待线程池中的任务完成执行
            if (!threadPool.awaitTermination(60, TimeUnit.SECONDS)) {
                // 调用 shutdownNow 取消正在执行的任务
                threadPool.shutdownNow();
                // 再次等待 60 s，如果还未结束，可以再次尝试，或则直接放弃
                if (!threadPool.awaitTermination(60, TimeUnit.SECONDS)) {
                    log.error("线程池任务未正常执行结束");
                }
            }
        } catch (InterruptedException ie) {
            // 捕获异常，重新调用 shutdownNow
            threadPool.shutdownNow();
        }
        // 任然没有关闭，循环关闭1000次，每次等待10毫秒
        if (!threadPool.isTerminated()) {
            try {
                for (int i = 0; i < 1000; i++) {
                    if (threadPool.awaitTermination(10, TimeUnit.MILLISECONDS)) {
                        break;
                    }
                    threadPool.shutdownNow();
                }
            } catch (Throwable e) {
                log.error(e.getMessage());
            }
        }
    }
}
